package com.seamtop.cuber.web.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.seamtop.cuber.common.base.DataObject;
import com.seamtop.cuber.common.metadata.Column;
import com.seamtop.cuber.common.metadata.TableMetaData;
import com.seamtop.cuber.common.metadata.ValueTypeContants;
import com.seamtop.cuber.core.ESManager;
import com.seamtop.cuber.core.HBaseUtil;
import com.seamtop.cuber.core.HqlAppendUtil;
import com.seamtop.cuber.web.service.CuberInitService;
import io.searchbox.core.Delete;
import io.searchbox.core.Search;
import io.searchbox.core.SearchResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * Created by zongjunfeng on 2015/9/6.
 */
@Service
public class CuberInitServiceImpl implements CuberInitService {

    private static final Logger LOG = LoggerFactory.getLogger(CuberInitServiceImpl.class);

    @Override
    public void createTable() {
        HashMap<String,TableMetaData> metaDataMap = DataObject.getMetaDataMap();
        for(String tableName:metaDataMap.keySet()){
            TableMetaData tableMetaData = metaDataMap.get(tableName);
            String table = tableName;
            if(tableMetaData.isVirtualTable()){
                table = tableMetaData.getPrimaryTableName();
            }
            boolean existFlag = true;
            try{
                existFlag = HBaseUtil.isTableExists(table);
            }catch (Exception e){
                e.printStackTrace();
            }

            HashMap<String,Column> columnMap = tableMetaData.getColumnMap();
            if(!existFlag){
                //创建HBase表结构

                List<String> familyList = new ArrayList<String>();
                for(String columnName:columnMap.keySet()){
                    String family = columnMap.get(columnName).getFamilyName();
                    if(!familyList.contains(family)){
                        familyList.add(family);
                    }
                }
                if(familyList.size() == 0){
                    familyList.add("mf");
                }
                try{
                    HBaseUtil.createTable(table,familyList);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            String hiveTableName = "Hive_" + tableName;
            try{
                HqlAppendUtil.dropTable(hiveTableName);//delete hive tables
            }catch (Exception e){
                e.printStackTrace();
            }

            //创建Hive表结构
            System.out.println("---------------------------------------------------------");
            StringBuilder sbr = new StringBuilder();
            sbr.append("CREATE EXTERNAL TABLE HIVE_"+tableName+"(");
            sbr.append("key string");
            int i=0;
            for(String columnName:columnMap.keySet()){
                sbr.append(",");
                i++;
                Column column = columnMap.get(columnName);
                sbr.append(column.getColumnName() + " " + ValueTypeContants.getHiveColumnTypeDesc(column));
            }
            sbr.append(")");
            sbr.append(" STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'");
            sbr.append(" WITH SERDEPROPERTIES");
            sbr.append(" (\"hbase.columns.mapping\" =");
            sbr.append(" \":key,");
            i=0;
            for(String columnName:columnMap.keySet()){
                i++;
                Column column = columnMap.get(columnName);
                sbr.append(column.getFamilyName()+":"+column.getMatch());
                if(column.isIncrement()){
                    sbr.append("#b");
                }
                if(i != columnMap.size()){
                    sbr.append(",");
                }
            }
            sbr.append("\")");
            sbr.append(" TBLPROPERTIES(\"hbase.table.name\" = \"" + table + "\")");
            LOG.info(" Create Hive Table[" + tableName + "] -- hiveSql:" + sbr.toString());
            try{
                HqlAppendUtil.executeSql(sbr.toString());
            }catch (Exception e){
                e.printStackTrace();
                LOG.error(" Create Hive Table[" + tableName + "] failed!",e);
            }
            LOG.info(" Create Hive Table[" + tableName + "] success!");
            System.out.println("---------------------------------------------------------");
        }
    }

    @Override
    public void coverTable() {
        deleteAndCreateTable();
    }

    @Override
    public void deleteAndCreateTable() {
        System.out.println("delete table ");
        HashMap<String,TableMetaData> metaDataMap = DataObject.getMetaDataMap();
        for(String s:metaDataMap.keySet()){
            TableMetaData metaData = metaDataMap.get(s);
            String hiveTableName = "Hive_" + s;
            String hbaseTableName = "";
            if(metaData.isVirtualTable()){
                hbaseTableName = metaData.getPrimaryTableName();
            }else {
                hbaseTableName = s;
            }
            try{
                HqlAppendUtil.dropTable(hiveTableName);//delete hive tables
                HBaseUtil.dropTable(hbaseTableName);//delete hbase tables
            }catch (Exception e){
                e.printStackTrace();
            }

//            //创建HBase表结构
            HashMap<String,Column> columnMap = metaData.getColumnMap();
            List<String> familyList = new ArrayList<String>();
            for(String columnName:columnMap.keySet()){
                String family = columnMap.get(columnName).getFamilyName();
                if(!familyList.contains(family)){
                    familyList.add(family);
                }
            }
            if(familyList.size() == 0){
                familyList.add("mf");
            }
            try{
                HBaseUtil.createTable(hbaseTableName,familyList);
            }catch (Exception e){
                e.printStackTrace();
            }

            //创建Hive表结构
            System.out.println("---------------------------------------------------------");
            StringBuilder sbr = new StringBuilder();
            sbr.append("CREATE EXTERNAL TABLE HIVE_"+s+"(");
            sbr.append("key string");
            int i=0;
            for(String columnName:columnMap.keySet()){
                sbr.append(",");
                i++;
                Column column = columnMap.get(columnName);
                //sbr.append(column.getColumnName() + " " + ValueTypeContants.getHiveColumnTypeDesc(column.getColumnType()));
                sbr.append(column.getColumnName() + " " + ValueTypeContants.getHiveColumnTypeDesc(column));
            }
            sbr.append(")");
            sbr.append(" STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'");
            sbr.append(" WITH SERDEPROPERTIES");
            sbr.append(" (\"hbase.columns.mapping\" =");
            sbr.append(" \":key,");
            i=0;
            for(String columnName:columnMap.keySet()){
                i++;
                Column column = columnMap.get(columnName);
                sbr.append(column.getFamilyName()+":"+column.getMatch());
                //int coumnType = column.getColumnType();
//                if(coumnType == 2 || coumnType == 3 || coumnType == 4 || coumnType == 5 ||coumnType == 7){
//                    sbr.append("#b");
//                }
                if(column.isIncrement()){
                    sbr.append("#b");
                }
                if(i != columnMap.size()){
                    sbr.append(",");
                }
            }
            sbr.append("\")");
            sbr.append(" TBLPROPERTIES(\"hbase.table.name\" = \"" + hbaseTableName + "\")");
            LOG.info(" Create Hive Table[" + s + "] -- hiveSql:" + sbr.toString());
            try{
                HqlAppendUtil.executeSql(sbr.toString());
            }catch (Exception e){
                e.printStackTrace();
                LOG.error(" Create Hive Table[" + s + "] failed!",e);
            }
            LOG.info(" Create Hive Table[" + s + "] success!");
            System.out.println("---------------------------------------------------------");


            //删除kafka数据
            try{
                ESManager.getClient().execute(new Delete.Builder("1")
                        .index("twitter")
                        .type("tweet")
                        .build());
            }catch (Exception e){
                e.printStackTrace();
            }


        }

    }


    public static void main(String [] args) throws Exception{
        String query = "{\"query\":{\"bool\":{\"must\":[{\"match_all\":{}}],\"must_not\":[],\"should\":[]}},\"from\":0,\"size\":10,\"sort\":[],\"facets\":{}}";
        Search search = new Search.Builder(query)
                // multiple index or types can be added.
                .addIndex("2sc")
                .addType("car_hit")
                .build();
        SearchResult result = ESManager.getClient().execute(search);
        String re = result.getJsonString();
        JSONObject object = JSON.parseObject(re);
        JSONArray o = object.getJSONObject("hits").getJSONArray("hits");
        for(int i=0;i<o.size();i++){
            JSONObject t = o.getJSONObject(i);
            ESManager.getClient().execute(new Delete.Builder(t.getString("_id"))
                    .index("2sc")
                    .type("car_hit")
                    .build());
        }
    }

}
